﻿
namespace Apache.Cassandra.Client {
    using System;
    using System.Diagnostics;
    using System.Net;

    internal class CassandraInternalConnection : IDisposable {
        private readonly WeakReference owner = new WeakReference(null, false);

        private Cassandra.Client client;
        private CassandraConnectionPool pool;
        private EndPoint endpoint;

        private readonly string initialKeyspace;

        private int pooledCount;
        private bool doomed;

        internal CassandraInternalConnection(Cassandra.Client client, EndPoint endpoint, CassandraConnectionString options) {
            this.client = client;
            this.endpoint = endpoint;

            initialKeyspace = options.InitialKeyspace;

            try {
                Connect();
            } catch (System.OutOfMemoryException) {
                DoomConnection();
                throw;
            } catch (System.StackOverflowException) {
                DoomConnection();
                throw;
            } catch (System.Threading.ThreadAbortException) {
                DoomConnection();
                throw;
            }
        }

        internal Cassandra.Client Client {
            get {
                return client;
            }
        }

        internal CassandraConnectionPool Pool {
            get {
                return pool;
            }
        }

        internal EndPoint EndPoint {
            get {
                return endpoint;
            }
        }

        internal bool CanBePooled {
            get {
                return !doomed && !owner.IsAlive;
            } 
        } 

        internal bool IsDoomed {
            get {
                return doomed;
            }
        }
        internal bool IsEmancipated { 
            get {
                return (pooledCount < 1) && !owner.IsAlive; 
            } 
        }

        internal void MakeNonPooledConnection(CassandraConnection owningObject) {
            owner.Target = owningObject;
            pooledCount = -1;
        }

        internal void MakePooledConnection(CassandraConnectionPool pool) {
            this.pool = pool;
        }

        internal void Activate() {
        }

        internal void Close(CassandraConnection owningObject) {
            Debug.Assert(null != owningObject, "null owning object");
            Debug.Assert(owningObject == owner.Target, "closing with a different owner");

            CassandraConnectionPool pool = Pool;
            if (null != pool) {
                pool.PushConnection(this, owningObject);
            } else {
                CassandraConnectionFactory.Factory.PerformanceCounters.NumberOfNonPooledConnections.Decrement();
                CassandraConnectionFactory.Factory.PerformanceCounters.HardDisconnectsPerSecond.Increment();
                Dispose();
            }
        }

        private void Disconnect() {
            client.OutputProtocol.Transport.Close();
            if (!Object.ReferenceEquals(client.InputProtocol.Transport, client.OutputProtocol.Transport)) {
                client.InputProtocol.Transport.Close();
            }
        }

        private void Connect() {
            try {
                client.InputProtocol.Transport.Open();
                if (!Object.ReferenceEquals(client.InputProtocol.Transport, client.OutputProtocol.Transport)) {
                    client.OutputProtocol.Transport.Open();
                }

                if (!String.IsNullOrEmpty(initialKeyspace)) {
                    client.set_keyspace(initialKeyspace);
                }
            } catch (Exception e) {
                if (e.IsCatchableType()) {
                    Disconnect();
                }
                throw;
            }
        }

        internal void Deactivate() {
            try {
                if (!IsDoomed) {
                    // TODO: Check client TTL once implemented as a pool option

                    if (!String.IsNullOrEmpty(initialKeyspace)) {
                        client.set_keyspace(initialKeyspace);
                    }
                }
            } catch (Exception e) {
                DoomConnection();

                if (!e.IsCatchableType()) {
                    throw;
                }
            }
        }

        internal void DoomConnection() {
            doomed = true;
        }

        internal void PrePush(CassandraConnection expectedOwner) { 
            if (null == expectedOwner) { 
                if (null != owner.Target) {
                    throw new ApplicationException("UnpooledObjectHasOwner");
                }
            } else if (owner.Target != expectedOwner) {
                throw new ApplicationException("UnpooledObjectHasWrongOwner");
            }
            if (0 != pooledCount) { 
                throw new ApplicationException("PushingObjectSecondTime");
            }
            pooledCount++; 
            owner.Target = null;
        } 
 
        internal void PostPop(object newOwner) {
            Debug.Assert(!IsEmancipated, "pooled object not in pool");
 
            if (null != owner.Target) { 
                throw new ApplicationException("PooledObjectHasOwner");
            } 
            owner.Target = newOwner;
            pooledCount--;
            if (null != Pool) {
                if (0 != pooledCount) { 
                    throw new ApplicationException("PooledObjectInPoolMoreThanOnce");
                }
            } else if (-1 != pooledCount) { 
                throw new ApplicationException("NonPooledObjectUsedMoreThanOnce");
            } 
        }

        public void Dispose() {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        private bool disposed = false;
        private void Dispose(bool disposing) {
            if (disposed) {
                return;
            }

            if (disposing) {
                Disconnect();
            }

            disposed = true;
        }
    }
}
